package com.lzzy

import org.apache.spark.sql._
import org.apache.spark.sql.expressions.Window
import java.util.Properties
import org.apache.spark.sql.functions._

object Term_Mysql_test {
  @throws[AnalysisException]
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("term_demo_scala").master("local").getOrCreate
    //写入数据库配置信息
    val url = "jdbc:mysql://127.0.0.1/term_demo?useSSL=false&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Chongqing"
    val prop = new Properties
    prop.setProperty("user", "root")
    prop.setProperty("password", "123456")
    //从数据库读取源数据
    val mysql_da: DataFrame = spark.read.jdbc(url, "foods", prop)
    //去到重复菜谱，为某些用不到的场景做铺垫
    val de_Data = mysql_da.dropDuplicates("菜名")
    de_Data.printSchema()
    val num = de_Data.count
    System.out.println("去掉重复菜谱后的数量为：" + num)
    //1 查询数据源里面属于一个作者的菜品和总收藏量有多少，保存前100个作者，存储下来
    val num_foods = de_Data.groupBy("作者id")
      .agg(functions.count("*")
        .alias("菜品数量"),
      functions.sum("收藏人数")
        .alias("总收藏数量"))
      .sort(functions.desc("菜品数量")).limit(100)
    //打印看看结果是否出来
    System.out.println("*************菜谱数量top100*********")
    //        num_foods.show();

    //存储至本地数据库
    num_foods.write.mode(SaveMode.Overwrite).jdbc(url, "foods_num", prop)
    //存储到hdfs
    num_foods.write.format("parquet")
      .option("header", "true")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("hdfs://20210322045-master:9000/term_data/foods_num")
    //2 查询数据里面不同年份最多收藏人数的前10菜品
    // 菜名|网址|作者id|收藏人数|月度最佳时间|...
    // 将最佳年月字段转换为年份
    System.out.println("做到第二题了")
    val de_year = de_Data.withColumnRenamed("收藏人数", "sl")
    de_year.createOrReplaceTempView("foods")
    spark.sql("SELECT *, CAST(SUBSTRING(`最佳年月`, 1, 4) AS int) as year FROM foods")
      .createOrReplaceTempView("foods_with_year")
    // 查询不同年份中收藏人数最多的前10道菜
    val year = spark.sql("SELECT * FROM (SELECT *, row_number() " +
      "OVER (PARTITION BY year ORDER BY sl desc ) AS rank_no FROM foods_with_year ) tmp WHERE rank_no <= 10 ")

    //分两步进行sql查询，第一步是开窗函数进行分组统计，第二步是根据年份和收藏人数排序
    year.createOrReplaceTempView("year_tab1")
    val foods_year = spark.sql("select `year`, `菜名`,`用料食材和数量`, `链接地址`, `作者id`, `sl`,`rank_no` " +
      "from year_tab1 order by `year` asc, `sl` desc")
    foods_year.show(100, false)
    //存储至本地数据库
    foods_year.write.mode(SaveMode.Overwrite).jdbc(url, "foods_year", prop)
    //存储到hdfs
    foods_year.write.format("parquet")
      .option("header", "true")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("hdfs://20210322045-master:9000/term_data/foods_year")
    //3 根据创建时间再添加一个字段：季节，比如3-5月是春季，6-8是夏季~
    //根据季节来进行分组计数，计算出每个季节收藏数量排名前5的菜品
    // 添加季节字段
    var data_jj = de_Data.withColumn("季节", functions.when(month(col("创建时间"))
      .between(3, 5), "春季").when(month(col("创建时间"))
      .between(6, 8), "夏季").when(month(col("创建时间"))
      .between(9, 11), "秋季").otherwise("冬季"))

    // 把季节改成英文方便开窗函数运行
    data_jj = data_jj.withColumnRenamed("季节", "season")
    data_jj = data_jj.withColumnRenamed("收藏人数", "sl")
    data_jj.createTempView("data_jj")
    val windowSpec = Window.partitionBy("season").orderBy(functions.desc("sl"))
    val data_jj1 = data_jj.withColumn("rank_no", row_number.over(windowSpec))
      .orderBy(expr("CASE season " +
        "WHEN '春季' THEN 1 " +
        "WHEN '夏季' THEN 2 " +
        "WHEN '秋季' THEN 3 " +
        "WHEN '冬季' THEN 4 " +
        "ELSE 5 " + "END"), col("rank_no"))
      .filter(col("rank_no").leq(5))
    System.out.println("*************每个季节收藏数量排名前5的菜品*********")

    //       将数据存储到本地数据库和hdfs集群
    //保存模式为覆盖
    data_jj1.write.mode(SaveMode.Overwrite).jdbc(url, "foods_season", prop)
    //存储到hdfs
    data_jj.write.format("parquet")
      .option("header", "true")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("hdfs://20210322045-master:9000/term_data/foods_season")

    //4每年收藏数量前3的作者和总收藏量
    mysql_da.createTempView("ws_data")
    spark.sql("SELECT `最佳年月`, `作者id`, `年收藏量`\n" +
      "FROM (\n" + "  SELECT `最佳年月`, `作者id`, SUM(`收藏人数`) AS `年收藏量`,\n" +
      "ROW_NUMBER() OVER(PARTITION BY FLOOR(`最佳年月` / 100) ORDER BY Max(`收藏人数`) DESC) AS `排名`\n" + "  " +
      "FROM ws_data\n" + "  GROUP BY `最佳年月`, `作者id`\n" + ") AS subquery\n" + "WHERE `排名` <= 3\n" + "ORDER BY `最佳年月`,`排名`")
      .createOrReplaceTempView("foods_with_year")

    val fsj = spark.sql("SELECT CAST(SUBSTRING(`最佳年月`, 1, 4) AS int) as `年份` ,`作者id`, `年收藏量` FROM foods_with_year")
    fsj.show()
    //存储至本地数据库
    fsj.write.mode(SaveMode.Overwrite).jdbc(url, "foods_nszl", prop)
    //存储到hdfs
    fsj.write.format("parquet")
      .option("header", "true")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("hdfs://20210322045-master:9000/term_data/foods_nscl")

    //5.每年的收藏率趋势
    mysql_da.createTempView("ws_data1")
    // 查询每个最佳年月的作者的年收藏量，并按照排名进行排序
    spark.sql("SELECT `最佳年月`, `作者id`, SUM(`收藏人数`) AS `年收藏量`,\n" + "" +
      "ROW_NUMBER() OVER(PARTITION BY FLOOR(`最佳年月` / 100) ORDER BY MAX(`收藏人数`) DESC) AS `排名`\n" +
      "FROM ws_data1\n" + "GROUP BY `最佳年月`, `作者id`\n" + "ORDER BY `最佳年月`,`排名`")
      .createOrReplaceTempView("foods_zly")
    // 提取年份、作者ID和年收藏量
    val zzl = spark.sql("SELECT CAST(SUBSTRING(`最佳年月`, 1, 4) AS int) AS `年份`, `作者id`, `年收藏量` FROM foods_zly")
    // 计算每年的总收藏人数增长趋势
    var trend = zzl.groupBy("`年份`")
      .agg(sum("`年收藏量`")
        .as("总收藏人数")).orderBy("`年份`")

    // 计算增长率
    val windowSpec1 = Window.orderBy("年份")
    trend = trend.withColumn("前一年收藏人数", lag("`总收藏人数`", 1).
      over(windowSpec1)).withColumn("增长率",
      round(expr("(cast(`总收藏人数` as double) / cast(`前一年收藏人数` as double)) - 1"), 2))
      .drop("前一年收藏人数")
    trend.show()
    trend.write.mode(SaveMode.Overwrite).jdbc(url, "foods_zzl", prop)
    //存储到hdfs
    trend.write.format("parquet")
      .option("header", "true").option("encoding", "UTF-8")
      .mode("overwrite")
      .save("hdfs://20210322045-master:9000/term_data/foods_zzl")
    spark.stop()
  }
}
